package io.gitee.hefren.hhmq.client;

import com.google.common.base.MoreObjects;
import com.google.common.collect.Lists;
import io.gitee.hefren.hhmq.model.HhMessage;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @Date 2024/7/21
 * @Author lifei
 */
public class HhMessageQueue {

    private final BlockingQueue<HhMessage<?>> queue = new LinkedBlockingQueue<>();
    private String topic;
    private List<HhMessageListener> listenerList = Lists.newArrayList();

    public HhMessageQueue(String topic) {
        this.topic = topic;
    }

    public boolean send(HhMessage<?> hhMessage) {
        boolean ok = queue.offer(hhMessage);
        if (ok && Objects.nonNull(listenerList) && !listenerList.isEmpty()) {
            listenerList.forEach(listener->listener.onMessage(hhMessage));
        }
        return ok;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public HhMessage<?> poll(long timeout) {
        try {
            return queue.poll(timeout, TimeUnit.MILLISECONDS);
        }catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public String toString() {
        return MoreObjects.toStringHelper(HhMessageQueue.class)
                .add("topic", topic)
                .toString();
    }

    public void addMessageListener(HhMessageListener hhMessageListener) {
        this.listenerList.add(hhMessageListener);
    }
}
